package com.globalwave.datasync.engine;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.lang.StringUtils;

import com.globalwave.common.C;
import com.globalwave.common.db.ConnectionProvider;
import com.globalwave.common.entity.SyncBatch;
import com.globalwave.common.entity.SyncRecordBackup;
import com.globalwave.common.service.SyncRecordBackupService;
import com.globalwave.common.service.SyncRecordBackupServiceImpl;
import com.globalwave.datasync.SyncColumn;
import com.globalwave.datasync.SyncConfigLoader;

public class OneByOneSync extends BaseSync {
	
	private SyncBatch batch ;
	private SyncRecordBackupService syncRecordBackupService ;
	private ConnectionProvider connectionProvider ;
	
	public OneByOneSync(ConnectionProvider connectionProvider) {
		this.connectionProvider = connectionProvider ;
	}
	
	@Override
	public void transfer() throws Exception {
		
		batch = (SyncBatch)context.get("batch") ;
		
		String sourceDbCode = (String)context.get("sourceDbCode") ;
		String syncConfigCode = batch.getSync_config_code() ;
		
		config = SyncConfigLoader.getInstance().getConfig(sourceDbCode, syncConfigCode) ;
		if (StringUtils.isEmpty(sourceDbCode)) {
			sourceDbCode = batch.getSource_db_code() ;
		}
		boolean isLog = config.isLog() ;

    	srcConnection = connectionProvider.getConnection(config.getSource().getDbCode()) ;
    	destConnection = connectionProvider.getConnection(config.getDestination().getDbCode()) ;

        Object[] params = getPkValues() ;
        
        ResultSet srcResultSet = 
            queryResultSet(
                    srcConnection, 
                    config.getSourceQuerySQL(), 
                    params) ;

        ResultSet targetResultSet = 
            queryResultSet(
                    destConnection, 
                    config.getDestinationQuerySQL(), 
                    params) ;

        boolean hasSrcRows = srcResultSet.next() ;
        List<Object[]> srcRecord = null ;
        if (hasSrcRows && isLog) {
        	srcRecord = getRecordMap(srcResultSet) ;
        }

        boolean hasTargetRows = targetResultSet.next() ;
        List<Object[]> destRecord = null ;
        if (hasTargetRows && isLog) {
        	destRecord = getRecordMap(targetResultSet) ;
        }
        
        String sql = null ;
        try {
	        destConnection.setAutoCommit(true) ;
	        
	        // 根据源数据记录和目标数据记录查询情况，确定具体操作
	        if (!hasSrcRows) {
	        	sql = config.getDestinationDeleteSQL() ;
	        	if (execute(destConnection, sql, params) > 1) {
	        		//throw new Exception("delete rows more than 1, so it is forbidden") ;
	        		log.warn("delete rows more than 1, so it is forbidden") ;
	        	}
	            //executeOne(config.getDestinationDeleteSQL(), null, params) ;
	        } else if (hasTargetRows) {
	        	sql = config.getDestinationUpdateSQL() ;
	        	PreparedStatement ps = getPrepareStatement(destConnection, sql);
	        	copyOneRowData(srcResultSet, ps) ;
	        	if (ps.executeUpdate() > 1) {
	        		//throw new Exception("update rows more than 1, so it is forbidden") ;
	        		log.warn("update rows more than 1, so it is forbidden") ;
	        	}
	            //executeOne(config.getDestinationUpdateSQL(), srcResultSet, params) ;
	        } else {
	        	sql = config.getDestinationInsertSQL() ;
	        	PreparedStatement ps = getPrepareStatement(destConnection, sql);
	        	copyOneRowData(srcResultSet, ps) ;
	        	if (ps.executeUpdate() > 1) {
	        		//throw new Exception("insert rows more than 1, so it is forbidden") ;
	        		log.warn("insert rows more than 1, so it is forbidden") ;
	        	}
	            //executeOne(config.getDestinationInsertSQL(), srcResultSet, null) ;
	        }

	        if (isLog) {
	        	Connection con = connectionProvider.getDefaultConnection() ;
	        	syncRecordBackupService = new SyncRecordBackupServiceImpl(con) ;
	        	recordBackup(srcRecord, true) ;
	        	recordBackup(destRecord, false) ;
	        }
        } catch (final Exception e) {
        	log.error("current sql is - " + sql) ;
        	throw e ;
        }
	}

	private Object[] getPkValues() {
		SyncColumn[] keys = config.getSource().getKeys() ;
		Object[] values = new Object[keys.length] ;
		
	    for(int i = 0 ; i < keys.length ; i ++) {
	    	try {
				values[i] = PropertyUtils.getProperty(batch, keys[i].getRefColumnsCode()) ;
			} catch (Exception e) {
				// do notiong
			}
	    }
	    
	    return values ;
	}
	

	/**
	 * Object[]{Key-ColumnName, Value-ColumnValue}
	 * @param resultSet
	 * @return
	 * @throws Exception
	 */
	public List<Object[]> getRecordMap(ResultSet resultSet) throws Exception {
		if (resultSet == null) {
			return null ;
		}
		

        ResultSetMetaData meta = resultSet.getMetaData();
        
        int relationLen = meta.getColumnCount() ;
		List<Object[]> record = new ArrayList<Object[]>(relationLen) ;
        
        for (int i = 1 ; i <= relationLen ; i ++) {

            record.add(new Object[]{meta.getColumnName(i), resultSet.getObject(i)}) ;
        }
		
		return record ;
	}
	
	private void recordBackup(List<Object[]> record, boolean isSource) {

		try {
			SyncRecordBackup backup = new SyncRecordBackup() ;
			
			backup.setContent(map2String(record)) ;
			backup.setIs_source(isSource?C.TRUE_INTEGER:C.FALSE_INTEGER) ;
			backup.setSync_batch_id(batch.getSync_batch_id()) ;
			backup.setSync_config_id(config.getSyncConfigId()) ;
			syncRecordBackupService.create(backup) ;
		} catch (Exception e) {
			log.error(record, e) ;
		}
	}
	
	private String map2String(List<Object[]> record) {
		if (record == null || record.size() == 0) {
			return "(no record)" ;
		}
		
		StringBuffer bf = new StringBuffer() ;
		
		int size = record.size() ;
		for (int i = 0 ; i < size ; i ++) {
			Object[] column = record.get(i) ;
			bf.append(column[0]) ;
			bf.append("=") ;
			bf.append(column[1]) ;
			bf.append(";") ;
		}
		
		return bf.length()> 2500?bf.substring(0, 2500):bf.toString() ;
	}
}
